package com.kingjw.windowsapi

import com.kingjw.sourceTest.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 *
 * @Package: com.kingjw.windowsapi
 * @ClassName: TimeWindows
 * @Author: 王坚伟
 * @CreateTime: 2022/1/18 19:09
 * @Description:
 */
object TimeWindows {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 这里需要设置为1，不然会因为线程问题导致update失败
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.getConfig.setAutoWatermarkInterval(5000)
//    val inputStream: DataStream[String] = env.readTextFile("input/sensor.txt")

    val inputStream: DataStream[String] = env.socketTextStream("10.0.3.207", 7777)
    val dataStream = inputStream
      .map( data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).trim.toLong, arr(2).trim.toDouble)
      } ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(5)) {
      override def extractTimestamp(element: SensorReading): Long = element.timestamp*1000L
    })

    val latetag: OutputTag[(String, Double, Long)] = new OutputTag[(String, Double, Long)]("late")

    // 每15秒统计一次，窗口内各传感器所有温度的最小值，以及最新的时间戳
    val resultStream = dataStream
      .map( data => (data.id, data.temperature, data.timestamp) )
      .keyBy(_._1)    // 按照二元组的第一个元素（id）分组
      .timeWindow(Time.seconds(15))
      .allowedLateness(Time.minutes(1))
      .sideOutputLateData(latetag)
      .reduce((curRes, newData) => (curRes._1, curRes._2.min(newData._2), newData._3) )


//    val value: DataStream[(String, Double)] = dataStream.map(r => (r.id, r.temperature))
//      .keyBy(_._1)
////      .window(EventTimeSessionWindows.withGap(Time.seconds(10)))
////      .timeWindow(Time.seconds(15),Time.seconds(5))
//      .countWindow(5)
//      .reduce((r1, r2) => (r1._1, r1._2.min(r2._2)))

    resultStream.getSideOutput(latetag).print("late")  //获取'late'标志的侧输出流
    resultStream.print("result")
    env.execute("windows test")
  }

}
class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[SensorReading] {

  val bound:Long = 60*1000
  override def checkAndGetNextWatermark(lastElement: SensorReading, extractedTimestamp: Long): Watermark = {
    if (lastElement.id == "sensor_1"){
      new Watermark(extractedTimestamp-bound)
    }else{
      null
    }
  }

  override def extractTimestamp(element: SensorReading, previousElementTimestamp: Long): Long = {
    element.timestamp
  }
}